package chan.project.smartlawdatamanage.elastic;

import chan.project.smartlawmodel.entity.Instrument;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.List;
@Component
public class SimpleCanalClientExample {

    //实现数据库与搜索引擎一致性

    @Resource
    private ElasticsearchClient elasticsearchClient;

    public void startListening() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "canal", "canal");
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe("smart_lawyer\\.instrument"); // 订阅instrument表的变化
            connector.rollback();

            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                } else {
                    processMessage(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
            }
        } finally {
            connector.disconnect();
        }
    }

    private void processMessage(List<Entry> entries) {
        for (Entry entry : entries) {
            if (entry.getEntryType() != EntryType.ROWDATA) continue;

            RowChange rowChage;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("Error parsing row change", e);
            }

            EventType eventType = rowChage.getEventType();
            if (!"instrument".equalsIgnoreCase(entry.getHeader().getTableName())) continue;

            for (RowData rowData : rowChage.getRowDatasList()) {
                Instrument instrument = new Instrument();

                fillInstrument(rowData, instrument);

                // 使用id作为文档ID
                IndexRequest<Instrument> indexRequest = IndexRequest.of(i -> i
                        .index("instrument_index")
                        .id(String.valueOf(instrument.getId()))
                        .document(instrument)
                );

                try {
                    IndexResponse response = elasticsearchClient.index(indexRequest);
                    System.out.println("Indexed document ID: " + response.id());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void fillInstrument(RowData rowData, Instrument instrument) {
        for (Column column : rowData.getAfterColumnsList()) {
            switch (column.getName()) {
                case "id":
                    instrument.setId(Long.parseLong(column.getValue()));
                    break;
                case "create_time":
                    try {
                        instrument.setCreateTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(column.getValue()));
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                    break;
                case "update_time":
                    try {
                        instrument.setUpdateTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(column.getValue()));
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                case "case_reason":
                    instrument.setCaseReason(column.getValue());
                    break;
                case "case_number":
                    instrument.setCaseNumber(column.getValue());
                    break;
                case "judgment_date":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        try {
                            // 使用 SimpleDateFormat 来解析日期字符串
                            SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
                            instrument.setJudgmentDate(formatter.parse(column.getValue()));
                        } catch (Exception e) {
                            System.err.println("日期解析失败: " + column.getValue());
                            e.printStackTrace();
                        }
                    }
                    break;
                case "title":
                    instrument.setTitle(column.getValue());
                    break;
                case "content":
                    instrument.setContent(column.getValue());
                    break;
                case "court":
                    instrument.setCourt(column.getValue());
                    break;
            }
        }
    }
}